package com.yunhe.history.util;

import com.yunhe.common.util.DateTimeUtils;
import com.yunhe.common.util.StringUtil;
import com.yunhe.history.constant.InfluxDBConsts;
import com.yunhe.common.constant.MeasurementConsts;
import com.yunhe.common.exception.ArgumentErrorException;
import com.yunhe.common.model.data.Measurement;
import com.yunhe.common.util.CollectionUtil;
import org.influxdb.InfluxDB;
import org.influxdb.dto.BatchPoints;
import org.influxdb.dto.Point;
import org.influxdb.dto.QueryResult;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;
import java.util.concurrent.TimeUnit;

/**
 * InfluxDB数据库工具类
 * @author liuronglei
 */
@Component
public class InfluxDBUtil {
	@Autowired
	private InfluxDBConnection influxDBConnection;
//	@Autowired
//	private AuthorityClient authorityClient;
//	@Autowired
//	private SimpleLocalCache simpleLocalCache;
	/**
	 * 时间倍数（用于毫秒转纳秒）
	 */
	private static final int timeMultiple = 1000000;
	/**
	 * 点号查询切片数量
	 */
	private static final int searchShardSize = 100;
	/**
	 * 量测插入切片数量
	 */
	private static final int insertShardSize = 10000;

	/**
	 * 根据点号获得时区ID
	 * @param pointNumber 点号
	 * @return String 时区ID
	 */
	public String getTimeZoneByPointNumber(Long pointNumber) {
		//TODO 因为目前电站还未配置code，所以先写死1108
		String pid = String.valueOf(pointNumber);
		if (pid != null && pid.startsWith("1108")) {
			return "Australia/Queensland";
		} else {
			return null;
		}

//		if (pid != null && pid.length() >= 4) {
//			// 取点号前4位作为项目code
//			String projectCode = pid.substring(0, 4);
//			String timeZoneId = (String) simpleLocalCache.getValue(projectCode);
//			if (timeZoneId == null) {
//				timeZoneId = authorityClient.getTimeZoneId(Long.parseLong(projectCode)).getBody().getResults();
//				if (timeZoneId == null) {
//					timeZoneId = "";
//				}
//				simpleLocalCache.setValue(projectCode, timeZoneId, CacheConsts.getDefaultTime());
//			}
//			return timeZoneId;
//		} else {
//			return null;
//		}
	}

	/**
	 * 填充rtime
	 * @param measurement 量测对象
	 * @param pointNumber 点号
	 */
	private void fillRtime(Measurement measurement, Long pointNumber) {
		if (measurement.getRtime() == null && !StringUtil.isEmpty(measurement.getDtime())) {
			String timeZoneId = getTimeZoneByPointNumber(pointNumber);
			measurement.setRtime(DateTimeUtils.getTimestamp(measurement.getDtime(), timeZoneId));
		}
	}

	/**
	 * 获得时间条件（纳秒级别的时间戳）
	 * @param dtime 时间条件字符串
	 * @param pointNumber 点号
	 * @return long 纳秒级别的时间戳
	 */
	private long getTimeCondition(String dtime, Long pointNumber) {
		String timeZoneId = getTimeZoneByPointNumber(pointNumber);
		return DateTimeUtils.getTimestamp(dtime, timeZoneId);
	}

	/**
	 * 根据influxDB返回的数据构建量测对象
	 * @param columns 字段
	 * @param values 值
	 * @return Measurement
	 */
	private Measurement convertMeasurement(List<String> columns, List<Object> values) {
		Measurement measurement = new Measurement();
		for(int i = 0; i < values.size(); i++) {
			String propertyName = setColumns(columns.get(i)); // 字段名
			Object value = values.get(i);
			if (value != null) {
				if (InfluxDBConsts.MEAS_COLUMN.R_TIME.value().equals(propertyName)) {
					measurement.setRtime(((Double)value).longValue());
				} else if (InfluxDBConsts.MEAS_COLUMN.POINT_NUMBER.value().equals(propertyName)) {
					measurement.setPointNumber(Long.parseLong((String)value));
				} else if (InfluxDBConsts.MEAS_COLUMN.VAL.value().equals(propertyName)) {
					measurement.setVal((Double)value);
				} else if (InfluxDBConsts.MEAS_COLUMN.FLAG.value().equals(propertyName)) {
					measurement.setFlag(((Double)value).intValue());
				}
			}
		}
		String timeZoneId = getTimeZoneByPointNumber(measurement.getPointNumber());
		measurement.setDtime(DateTimeUtils.getDateString(measurement.getRtime(), timeZoneId));
		return measurement;
	}

	/**
	 * 查询量测列表
	 * @param sql 查询字符串
	 * @return List<Measurement>
	 */
	private List<Measurement> getQueryData(String sql) {
		QueryResult results = influxDBConnection.query(sql);
		if (results.getResults().get(0).getSeries() == null) {
			return new ArrayList<>();
		}
		List<QueryResult.Series> series = results.getResults().get(0).getSeries();
		List<Measurement> lists = new ArrayList<>();
		for (QueryResult.Series serie : series) {
			List<String> columns = serie.getColumns();
			List<List<Object>> values = serie.getValues();
			if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
				return new ArrayList<>();
			}
			for (List<Object> list : values) {
				Measurement measurement = convertMeasurement(columns, list);
				if (measurement != null && measurement.getPointNumber() != null) {
					lists.add(measurement);
				}
			}
		}
		return lists;
	}

	/**
	 * 查询单条量测
	 * @param sql 查询字符串
	 * @return Measurement
	 */
	private Measurement getQueryDataFirst(String sql) {
		QueryResult results = influxDBConnection.query(sql);
		if (results.getResults().get(0).getSeries() == null) {
			return null;
		}
		QueryResult.Series serie = results.getResults().get(0).getSeries().get(0);
		List<String> columns = serie.getColumns();
		List<List<Object>> values = serie.getValues();
		if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
			return null;
		}
		List<Object> list = values.get(0);
		return convertMeasurement(columns, list);
	}

	/**
	 * 数据转换为结果对象
	 * @param columns 字段
	 * @param values 值
	 * @return Map<Long, Measurement>
	 */
	private Map<Long, Measurement> getQueryDataMap(List<String> columns, List<List<Object>> values) {
		if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
			return Collections.emptyMap();
		}
		Map<Long, Measurement> resultMap = new HashMap<>();
		for (List<Object> list : values) {
			Measurement measurement = convertMeasurement(columns, list);
			resultMap.put(measurement.getPointNumber(), measurement);
		}
		return resultMap;
	}

	/**
	 * 字段转义
	 * @param column 字段
	 * @return String
	 */
	private String setColumns(String column){
		String[] cols = column.split("_");
		StringBuffer sb = new StringBuffer();
		for(int i=0; i< cols.length; i++){
			String col = cols[i].toLowerCase();
			if(i != 0) {
				String start = col.substring(0, 1).toUpperCase();
				String end = col.substring(1).toLowerCase();
				col = start + end;
			}
			sb.append(col);
		}
		return sb.toString();
	}

	private String getPidCondition(List<Long> pointNumbers) {
        StringBuilder pointNumberStr = new StringBuilder();
		if (pointNumbers.size() == 1) {
            pointNumberStr.append(String.format("pid='%s'", pointNumbers.get(0)));
		} else {
            pointNumberStr.append("(");
			for (int i = 0; i < pointNumbers.size(); i++) {
				Long pointNumber = pointNumbers.get(i);
				if (i == 0) {
					pointNumberStr.append("pid='").append(pointNumber).append("'");
				} else {
					pointNumberStr.append(" or pid='").append(pointNumber).append("'");
				}
			}
			pointNumberStr.append(") ");
		}
        return pointNumberStr.toString();
	}

	/**
	 * 获得点号在时间区间内的量测总数量
	 * @param pointNumber 点号
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Long getCount(Long pointNumber,
                         List<String> dtime,
                         String frequency,
                         Integer flag) {
		List<Long> dtimeLong = new ArrayList<>();
		for (String myDtime : dtime) {
			dtimeLong.add(getTimeCondition(myDtime, pointNumber));
		}
		return getCountMilli(pointNumber, dtimeLong, frequency, flag);
	}

	/**
	 * 获得点号列表在时间区间内的量测总数量
	 * @param pointNumbers 点号列表
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 */
	public Long getCount(List<Long> pointNumbers,
                         List<String> dtime,
                         String frequency,
                         Integer flag) {
		List<Long> dtimeLong = new ArrayList<>();
		for (String myDtime : dtime) {
			dtimeLong.add(getTimeCondition(myDtime, pointNumbers.get(0)));
		}
		return getCountMilli(pointNumbers, dtimeLong, frequency, flag);
	}

	/**
	 * 获得点号在时间区间内的量测总数量
	 * @param pointNumber 点号
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Long getCountMilli(Long pointNumber,
                              List<Long> dtime,
                              String frequency,
                              Integer flag) {
        String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select count(val) from %s where 1=1", tableName));
        Optional<Long> pointNumberOpt = Optional.ofNullable(pointNumber);
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        pointNumberOpt.ifPresent(e->sqlBuilder.append(String.format(" and pid='%s'", e)));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		if (dtime != null && dtime.size() == 1) {
			Optional<Long> dateOpt = Optional.ofNullable(dtime.get(0));
			dateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time=%s", e * timeMultiple)));
		} else if (dtime != null && dtime.size() == 2) {
			Optional<Long> startDateOpt = Optional.ofNullable(dtime.get(0));
			startDateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time>=%s", e * timeMultiple)));
			Optional<Long> endDateOpt = Optional.ofNullable(dtime.get(1));
			endDateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time<=%s", e * timeMultiple)));
			sqlBuilder.append(" order by time asc");
		}
		QueryResult results = influxDBConnection.query(sqlBuilder.toString());
		if (results.getResults().get(0).getSeries() == null) {
			return null;
		}
		QueryResult.Series serie = results.getResults().get(0).getSeries().get(0);
		List<String> columns = serie.getColumns();
		List<List<Object>> values = serie.getValues();
		if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
			return null;
		}
		List<Object> list = values.get(0);
		if (list != null && list.size() > 1) {
			return ((Double)list.get(1)).longValue();
		} else {
			return 0L;
		}
	}

	/**
	 * 获得点号列表在时间区间内的量测总数量
	 * @param pointNumbers 点号列表
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 */
	public Long getCountMilli(List<Long> pointNumbers,
                              List<Long> dtime,
                              String frequency,
                              Integer flag) {
        String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select count(val) from %s where 1=1", tableName));
		Optional<List<Long>> pointNumberOpt = Optional.ofNullable(pointNumbers);
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
		pointNumberOpt.ifPresent(e->sqlBuilder.append(String.format(" and %s", getPidCondition(e))));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		if (dtime != null && dtime.size() == 1) {
			Optional<Long> dateOpt = Optional.ofNullable(dtime.get(0));
			dateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time=%s", e * timeMultiple)));
		} else if (dtime != null && dtime.size() > 1) {
			Optional<Long> startDateOpt = Optional.ofNullable(dtime.get(0));
			startDateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time>=%s", e * timeMultiple)));
			Optional<Long> endDateOpt = Optional.ofNullable(dtime.get(1));
			endDateOpt.ifPresent(e->sqlBuilder.append(String.format(" and time<=%s", e * timeMultiple)));
			sqlBuilder.append(" order by time asc");
		} else {
			return 0L;
		}
		QueryResult results = influxDBConnection.query(sqlBuilder.toString());
		if (results.getResults().get(0).getSeries() == null) {
			return 0L;
		}
		QueryResult.Series serie = results.getResults().get(0).getSeries().get(0);
		List<String> columns = serie.getColumns();
		List<List<Object>> values = serie.getValues();
		if (columns == null || values == null || columns.size() == 0 || values.size() == 0) {
			return 0L;
		}
		List<Object> list = values.get(0);
		if (list != null && list.size() > 1) {
			return ((Double)list.get(1)).longValue();
		} else {
			return 0L;
		}
	}

	/**
	 * 获得点号在时间区间内的所有量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultScanner(Long pointNumber,
                                              String startDate,
                                              String endDate,
                                              String frequency,
                                              Integer flag) {
		return getResultScannerMilli(pointNumber, getTimeCondition(startDate, pointNumber),
                getTimeCondition(endDate, pointNumber), frequency, flag);
	}

	/**
	 * 获得点号在时间区间内的所有量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultScannerMilli(Long pointNumber,
                                                   Long startDate,
                                                   Long endDate,
                                                   String frequency,
                                                   Integer flag) {
		if (pointNumber == null || startDate == null || endDate == null) {
			return new ArrayList<>();
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time>=%s and time<=%s",
                tableName, pointNumber, startDate * timeMultiple, endDate * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(" order by time asc");
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 获得点号在时间区间内的所有量测（批量点号）
	 * @param pointNumbers 点号列表
	 * @param startDate 开始时间
	 * @param endDate 结束时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultsScanner(List<Long> pointNumbers,
											   String startDate,
											   String endDate,
											   String frequency,
											   Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers) || startDate == null || endDate == null) {
			return new ArrayList<>();
		}
		// 用第一个点号来区分时区
		Long pointNumber = pointNumbers.get(0);
		// 如果点号超过一定数量，会出现效率问题，所以采用分片形式
		if (pointNumbers.size() > searchShardSize) {
			List<Measurement> result = new ArrayList<>();
			int size = pointNumbers.size();
			int index = size / searchShardSize;
			for (int i = 0; i < index + 1; i++) {
				int subEnd = size;
				if (i < index) {
					subEnd = i * searchShardSize + searchShardSize;
				}
				List<Measurement> measurements = getResultsScannerMilli(pointNumbers.subList(i * searchShardSize, subEnd),
						getTimeCondition(startDate, pointNumber), getTimeCondition(endDate, pointNumber), frequency, flag);
				if (measurements != null) {
					result.addAll(measurements);
				}
			}
			return result;
		} else {
			return getResultsScannerMilli(pointNumbers, getTimeCondition(startDate, pointNumber),
					getTimeCondition(endDate, pointNumber), frequency, flag);
		}
	}

	/**
	 * 获得点号在时间区间内的所有量测（批量点号）
	 * @param pointNumbers 点号列表
	 * @param startDate 开始时间
	 * @param endDate 结束时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultsScannerMilli(List<Long> pointNumbers,
													Long startDate,
													Long endDate,
													String frequency,
													Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers) || startDate == null || endDate == null) {
			return new ArrayList<>();
		}
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		String pidCondition = getPidCondition(pointNumbers);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where %s and time>=%s and time<=%s",
				tableName, pidCondition, startDate * timeMultiple, endDate * timeMultiple));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" order by time asc");
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 获得从开始时间后的所有量测列表
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultByStartDate(Long pointNumber,
												  String startDate,
												  String frequency,
												  Integer flag) {
		return getResultByStartDateMilli(pointNumber, getTimeCondition(startDate, pointNumber), frequency, flag);
	}

	/**
	 * 获得从开始时间后的所有量测列表
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return List<Measurement>
	 */
	public List<Measurement> getResultByStartDateMilli(Long pointNumber,
													   Long startDate,
													   String frequency,
													   Integer flag) {
		if (pointNumber == null || startDate == null) {
			return new ArrayList<>();
		}
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time>=%s",
				tableName, pointNumber, startDate * timeMultiple));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" order by time asc");
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 获得点号在时间区间内的第一个量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Measurement getResultScannerFirst(Long pointNumber,
                                             String startDate,
                                             String endDate,
                                             String frequency,
                                             Integer flag) {
		return getResultScannerFirstMilli(pointNumber, getTimeCondition(startDate, pointNumber),
                getTimeCondition(endDate, pointNumber), frequency, flag);
	}

	/**
	 * 获得点号在时间区间内的第一个量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Measurement getResultScannerFirstMilli(Long pointNumber,
                                                  Long startDate,
                                                  Long endDate,
                                                  String frequency,
                                                  Integer flag) {
		if (pointNumber == null || startDate == null || endDate == null) {
			return null;
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time>=%s and time<=%s",
                tableName, pointNumber, startDate * timeMultiple, endDate * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(" order by time asc limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	/**
	 * 获得点号在时间区间内的最后一个量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
	 * @return Measurement
	 */
	public Measurement getResultScannerLast(Long pointNumber,
                                            String startDate,
                                            String endDate,
                                            String frequency,
                                            Integer flag) {
		return getResultScannerLastMilli(pointNumber, getTimeCondition(startDate, pointNumber),
                getTimeCondition(endDate, pointNumber), frequency, flag);
	}

	/**
	 * 获得点号在时间区间内的最后一个量测
	 * @param pointNumber 点号
	 * @param startDate 开始时间
	 * @param endDate 结束时间
	 * @return Measurement
	 */
	public Measurement getResultScannerLastMilli(Long pointNumber,
                                                 Long startDate,
                                                 Long endDate,
                                                 String frequency,
                                                 Integer flag) {
		if (pointNumber == null || startDate == null || endDate == null) {
			return null;
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time>=%s and time<=%s",
                tableName, pointNumber, startDate * timeMultiple, endDate * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(" order by time desc limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	/**
	 * 获得点号在某个时间点的量测
	 * @param pointNumber 点号
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Measurement getRowData(Long pointNumber,
                                  String dtime,
                                  String frequency,
                                  Integer flag) {
		return getRowDataMilli(pointNumber, getTimeCondition(dtime, pointNumber), frequency, flag);
	}

	/**
	 * 获得点号在某个时间点的量测
     * 对于原始值的查询来说，查询≤该时间点的最后的值
	 * 对于非原始值的查询，则仅查询该时间点的值
	 * @param pointNumber 点号
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public Measurement getRowDataMilli(Long pointNumber,
                                       Long dtime,
                                       String frequency,
                                       Integer flag) {
		if (pointNumber == null || dtime == null) {
			return null;
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder();
        if (!MeasurementConsts.FREQUENCY.Original.equals(frequency)) {
			sqlBuilder.append(String.format("select * from %s where pid='%s' and time=%s",
					tableName, pointNumber, dtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		} else {
			sqlBuilder.append(String.format("select * from %s where pid='%s' and time<=%s",
					tableName, pointNumber, dtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
			sqlBuilder.append(" order by time desc");
		}
        sqlBuilder.append(" limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	/**
	 * 获得点号列表在某个时间点的量测
	 * @param pointNumbers 点号列表
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public List<Measurement> getRowData(List<Long> pointNumbers,
                                        String dtime,
                                        String frequency,
                                        Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers)) {
			return new ArrayList<>();
		}
		// 如果点号超过一定数量，会出现效率问题，所以采用分片形式
		if (pointNumbers.size() > searchShardSize) {
			List<Measurement> result = new ArrayList<>();
			int size = pointNumbers.size();
			int index = size / searchShardSize;
			for (int i = 0; i <= index; i++) {
				int subStart = i * searchShardSize;
				int subEnd = size;
				if (i < index) {
					subEnd = subStart + searchShardSize;
				}
				List<Measurement> measurements = getRowDataMilli(pointNumbers.subList(subStart, subEnd),
						getTimeCondition(dtime, pointNumbers.get(0)), frequency, flag);
				if (measurements != null) {
					result.addAll(measurements);
				}
			}
			return result;
		} else {
			return getRowDataMilli(pointNumbers, getTimeCondition(dtime, pointNumbers.get(0)), frequency, flag);
		}
	}

    /**
     * 获得点号列表在某个时间点的量测
     * @param pointNumbers 点号列表
     * @param rtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     * @return Measurement
     */
    public List<Measurement> getRowDataMilli(List<Long> pointNumbers,
                                             Long rtime,
                                             String frequency,
                                             Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers) || rtime == null) {
			return new ArrayList<>();
		}
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		// 拼装pid条件
		String pidCondition = getPidCondition(pointNumbers);
		StringBuilder sqlBuilder = new StringBuilder();
		if (!MeasurementConsts.FREQUENCY.Original.equals(frequency)) {
			sqlBuilder.append(String.format("select * from %s where %s and time=%s",
					tableName, pidCondition, rtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
			return getQueryData(sqlBuilder.toString());
		} else {
			sqlBuilder.append(String.format("select pid,last(val) as val,flag from %s where %s and time<=%s",
					tableName, pidCondition, rtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
			sqlBuilder.append(" group by pid");
			List<Measurement> result = getQueryData(sqlBuilder.toString());
			// 对于原始值，需要重新设置时间
			// 因为重复执行了时间格式化，此处稍微有些性能消耗，虽然影响不大，但后续可以再优化
			if (CollectionUtil.isNotEmpty(result)) {
				for (Measurement measurement : result) {
					String timeZoneId = getTimeZoneByPointNumber(measurement.getPointNumber());
					String dtime = DateTimeUtils.getDateString(rtime, timeZoneId);
					measurement.setRtime(rtime);
					measurement.setDtime(dtime);
				}
			}
			return result;
		}
    }

	/**
	 * 获得点号列表在某个时间点的量测（分页）
	 * @param pointNumbers 点号列表
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param page 页码
     * @param size 每页数量
	 * @return Measurement
	 */
	public List<Measurement> getRowData(List<Long> pointNumbers,
                                        String dtime,
                                        String frequency,
                                        Integer flag,
                                        int page,
                                        int size) {
		if (CollectionUtil.isEmpty(pointNumbers)) {
			return new ArrayList<>();
		}
		return getRowDataMilli(pointNumbers, getTimeCondition(dtime, pointNumbers.get(0)), frequency, flag, page, size);
	}

	/**
	 * 获得点号列表在某个时间范围的量测
	 * @param pointNumbers 点号列表
	 * @param dtime 时间范围
     * @param frequency 时间频率
     * @param flag 标记
	 * @return Measurement
	 */
	public List<Measurement> getRowData(List<Long> pointNumbers,
                                        List<String> dtime,
                                        String frequency,
                                        Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers)) {
			return new ArrayList<>();
		}
		return getRowDataMilliDesc(pointNumbers, getTimeCondition(dtime.get(0), pointNumbers.get(0)),
                getTimeCondition(dtime.get(1), pointNumbers.get(0)), frequency, flag);
	}

	/**
	 * 获得点号列表在某个时间范围的量测（分页）
	 * @param pointNumbers 点号列表
	 * @param dtime 时间范围
     * @param frequency 时间频率
     * @param flag 标记
     * @param page 页码
     * @param size 每页数量
	 * @return Measurement
	 */
	public List<Measurement> getRowData(List<Long> pointNumbers,
                                        List<String> dtime,
                                        String frequency,
                                        Integer flag,
                                        int page, int size) {
		if (CollectionUtil.isEmpty(pointNumbers) || CollectionUtil.isEmpty(dtime) || dtime.size() != 2) {
			return new ArrayList<>();
		}
		return getRowDataMilli(pointNumbers, getTimeCondition(dtime.get(0), pointNumbers.get(0)),
				getTimeCondition(dtime.get(1), pointNumbers.get(0)), frequency, flag, page, size);
	}

    /**
     * 获得点号列表在某个时间点的量测
     * @param pointNumbers 点号列表
     * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     * @return Measurement
     */
	public List<Measurement> getRowDataMilliDesc(List<Long> pointNumbers,
                                                 Long dtime,
                                                 String frequency,
                                                 Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers) || dtime == null) {
			return new ArrayList<>();
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
		// 拼装pid条件
		String pidCondition = getPidCondition(pointNumbers);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where %s and time=%s",
                tableName, pidCondition, dtime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(" order by time desc");
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 获得点号列表在某个时间点的量测
	 * @param pointNumbers 点号列表
	 * @param dtime 时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param page 页码
     * @param size 每页数量
	 * @return Measurement
	 */
	public List<Measurement> getRowDataMilli(List<Long> pointNumbers,
                                             Long dtime,
                                             String frequency,
                                             Integer flag,
                                             int page,
                                             int size) {
		if (CollectionUtil.isEmpty(pointNumbers) || dtime == null) {
			return new ArrayList<>();
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
		// 拼装pid条件
		String pidCondition = getPidCondition(pointNumbers);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where %s and time=%s",
                tableName, pidCondition, dtime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(String.format(" order by time desc LIMIT %s OFFSET %s", size, (page - 1) * size));
		return getQueryData(sqlBuilder.toString());
	}

	public List<Measurement> getRowDataMilliDesc(List<Long> pointNumbers,
                                                 Long startTime,
                                                 Long endTime,
                                                 String frequency,
                                                 Integer flag) {
		if (CollectionUtil.isEmpty(pointNumbers) || startTime == null || endTime == null) {
			return new ArrayList<>();
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
		// 拼装pid条件
		String pidCondition = getPidCondition(pointNumbers);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where %s and time>=%s and time<=%s",
                tableName, pidCondition, startTime * timeMultiple, endTime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(" order by time desc");
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 获得点号列表在某个时间范围的量测（分页）
	 * @param pointNumbers 点号列表
	 * @param startTime 开始时间
	 * @param endTime 结束时间
     * @param frequency 时间频率
     * @param flag 标记
     * @param page 页码
     * @param size 每页数量
	 */
	public List<Measurement> getRowDataMilli(List<Long> pointNumbers,
                                             Long startTime,
                                             Long endTime,
                                             String frequency,
                                             Integer flag,
                                             int page,
                                             int size) {
		if (CollectionUtil.isEmpty(pointNumbers)) {
			return new ArrayList<>();
		}
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
		// 拼装pid条件
		String pidCondition = getPidCondition(pointNumbers);
        StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where %s and time>=%s and time<=%s",
                tableName, pidCondition, startTime * timeMultiple, endTime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
        sqlBuilder.append(String.format(" order by time desc LIMIT %s OFFSET %s", size, (page - 1) * size));
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 批量获得点号最后一个量测（按时间排序，取最后一个）
	 * @param pointNumbers 点号
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return Measurement
	 */
	public Map<Long, Measurement> getResultScannerLast(List<Long> pointNumbers,
													   String frequency,
													   Integer flag,
													   String endTime) {
		if (CollectionUtil.isEmpty(pointNumbers)) {
			return new HashMap<>();
		}
		if (pointNumbers.size() > searchShardSize) {
			Map<Long, Measurement> result = new HashMap<>();
			int size = pointNumbers.size();
			int index = size / searchShardSize;
			for (int i = 0; i < index + 1; i++) {
				int subEnd = size;
				if (i < index) {
					subEnd = i * searchShardSize + searchShardSize;
				}
				Map<Long, Measurement> map = getResultScannerLastBatch(pointNumbers.subList(i * searchShardSize, subEnd),
						frequency, flag, endTime);
				if (map != null) {
					result.putAll(map);
				}
			}
			return result;
		} else {
			return getResultScannerLastBatch(pointNumbers, frequency, flag, endTime);
		}
	}

	/**
	 * 批量获得点号最后一个量测（按时间排序，取最后一个）
	 * @param pointNumbers 点号
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return Measurement
	 */
	private Map<Long, Measurement> getResultScannerLastBatch(List<Long> pointNumbers,
															 String frequency,
															 Integer flag,
															 String endTime) {
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		String pidCondition = getPidCondition(pointNumbers);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select time, pid, last(val) as val, flag from %s where %s",
				tableName, pidCondition));
		if (endTime != null) {
			sqlBuilder.append(String.format(" and time<=%s", getTimeCondition(endTime, pointNumbers.get(0)) * timeMultiple));
		}
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" group by pid");
		QueryResult results = influxDBConnection.query(sqlBuilder.toString());
		if (results.getResults().get(0).getSeries() == null) {
			return null;
		}
		Map<Long, Measurement> result = new HashMap<>();
		List<QueryResult.Series> series = results.getResults().get(0).getSeries();
		for (QueryResult.Series serie : series) {
			List<String> columns = serie.getColumns();
			List<List<Object>> values = serie.getValues();
			result.putAll(getQueryDataMap(columns, values));
		}
		return result;
	}

	/**
	 * 获得点号最后一个量测（按时间排序，取最后一个）
	 * @param pointNumber 点号
	 * @return Measurement
	 */
	public Measurement getResultScannerLast(Long pointNumber) {
		String tableName = InfluxDBConsts.getTableName(MeasurementConsts.FREQUENCY.Original.value());
		String sql = String.format("select time, pid, last(val) as val, flag from %s where pid='%s'",
				tableName, pointNumber);
		return getQueryDataFirst(sql);
	}

	/**
	 * 获得与传入值不同的上一个量测对象
	 * @param pointNumber 点号
	 * @param currentVal 传入值
	 * @return Measurement
	 */
	public Measurement getPreValue(Long pointNumber, Double currentVal) {
		String tableName = InfluxDBConsts.getTableName(MeasurementConsts.FREQUENCY.Original.value());
		String sql = String.format("select * from %s where pid='%s' and val<>%s order by time desc limit 1",
				tableName, pointNumber, currentVal);
		return getQueryDataFirst(sql);
	}

	public Measurement getPreValueByTime(Long pointNumber,
										 String dtime,
										 String frequency,
										 Integer flag) {
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		long time = getTimeCondition(dtime, pointNumber);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time<%s",
				tableName, pointNumber, time * timeMultiple));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" order by time desc limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	public Measurement getPreValueByRtime(Long pointNumber,
										  Long rtime,
										  String frequency,
										  Integer flag) {
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time<%s",
				tableName, pointNumber, rtime * timeMultiple));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" order by time desc limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	public Measurement getNextValueByTime(Long pointNumber,
										  String dtime,
										  String frequency,
										  Integer flag) {
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		long time = getTimeCondition(dtime, pointNumber);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select * from %s where pid='%s' and time>%s",
				tableName, pointNumber, time * timeMultiple));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" order by time asc limit 1");
		return getQueryDataFirst(sqlBuilder.toString());
	}

	/**
	 * 为表添加 or 更新数据
	 * @param measurement 量测对象
     * @param frequency 时间频率
	 */
	public void putData(Measurement measurement,
                        String frequency) {
		if (measurement == null || measurement.getPointNumber() == null
				|| measurement.getVal() == null || Double.isNaN(measurement.getVal())) {
			return;
		}
        String tableName = InfluxDBConsts.getTableName(frequency);
		Map<String, String> tags = new HashMap<>();
		String pointNumber = String.valueOf(measurement.getPointNumber());
		tags.put("pid", pointNumber);
		Map<String, Object> fields = new HashMap<>();
		// InfluxDB的字段类型，由第一次插入的值得类型决定
		fields.put("val", measurement.getVal());
		fields.put("flag", measurement.getFlag());
		fillRtime(measurement, measurement.getPointNumber());
		// 时间使用毫秒为单位
		influxDBConnection.insert(tableName, tags, fields, measurement.getRtime(), TimeUnit.MILLISECONDS);
	}

	/**
	 * 为表添加 or 更新数据（分片）
	 * @param measList 量测对象列表
     * @param frequency 时间频率
	 */
	public void putData(List<Measurement> measList,
                        String frequency) {
		if (measList.size() > insertShardSize) {
			int size = measList.size();
			int index = size / insertShardSize;
			for (int i = 0; i <= index; i++) {
				int subStart = i * insertShardSize;
				int subEnd = size;
				if (i < index) {
					subEnd = subStart + insertShardSize;
				}
				batchInsert(measList.subList(subStart, subEnd), frequency);
			}
		} else {
			batchInsert(measList, frequency);
		}
	}

	/**
	 * 为表添加 or 更新数据（批量）
	 * @param measList 量测对象列表
	 * @param frequency 时间频率
	 */
	private void batchInsert(List<Measurement> measList,
							 String frequency) {
		String tableName = InfluxDBConsts.getTableName(frequency);
		List<String> records = new ArrayList<>();
		for (Measurement item : measList) {
			if (item.getPointNumber() == null || item.getVal() == null || Double.isNaN(item.getVal())) { // || item.getFlag() == null
				continue;
			}
			Map<String, String> tags = new HashMap<>();
			String pointNumber = String.valueOf(item.getPointNumber());
			tags.put("pid", pointNumber);
			Map<String, Object> fields = new HashMap<>();
			fields.put("val", item.getVal());
			fields.put("flag", item.getFlag());
			fillRtime(item, item.getPointNumber());
			Point point = influxDBConnection.pointBuilder(tableName, item.getRtime(), tags, fields);
			BatchPoints batchPoints = BatchPoints.database(InfluxDBConsts.DATABASE).tag("pid", pointNumber)
					.retentionPolicy(InfluxDBConsts.RETENTION_POLICY).consistency(InfluxDB.ConsistencyLevel.ALL).build();
			batchPoints.point(point);
			records.add(batchPoints.lineProtocol());
		}
		influxDBConnection.batchInsert(InfluxDBConsts.DATABASE, InfluxDBConsts.RETENTION_POLICY, InfluxDB.ConsistencyLevel.ALL, records);
	}

	/**
	 * 根据点号和时间删除指定的行
	 * @param pointNumber 点号
	 * @param dtime 时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 * @return boolean
	 */
	public boolean deleteRow(Long pointNumber,
							 String dtime,
							 String frequency,
							 Integer flag) {
		return deleteRowMilli(pointNumber, getTimeCondition(dtime, pointNumber), frequency, flag);
	}

	/**
	 * 根据点号和时间删除指定的行
	 * @param pointNumber 点号
	 * @param dtime 时间
	 * @return boolean
	 */
	public boolean deleteRowMilli(Long pointNumber,
								  Long dtime,
								  String frequency,
								  Integer flag) {
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
        String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder(String.format("delete from %s where pid='%s' and time=%s",
                tableName, pointNumber, dtime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		String error = influxDBConnection.deleteMeasurementData(sqlBuilder.toString());
		return StringUtil.isEmpty(error);
	}

	/**
	 * 根据点号和时间范围删除指定的行
	 * @param pointNumber 点号
	 * @param startTime 开始时间
	 * @param endTime 结束时间
	 * @return boolean
	 */
	public boolean deleteRow(Long pointNumber,
                             String startTime,
                             String endTime,
							 String frequency,
							 Integer flag) {
		return deleteRowMilli(pointNumber, getTimeCondition(startTime, pointNumber),
                getTimeCondition(endTime, pointNumber), frequency, flag);
	}

	/**
	 * 根据点号和时间范围删除指定的行
	 * @param pointNumber 点号
	 * @param startTime 开始时间
	 * @param endTime 结束时间
	 * @return boolean
	 */
	public boolean deleteRowMilli(Long pointNumber,
                                  Long startTime,
                                  Long endTime,
                                  String frequency,
                                  Integer flag) {
        Optional<Integer> flagOpt = Optional.ofNullable(flag);
	    String tableName = InfluxDBConsts.getTableName(frequency);
        StringBuilder sqlBuilder = new StringBuilder(String.format("delete from %s where pid='%s' and time>=%s and time<=%s",
                tableName, pointNumber, startTime * timeMultiple, endTime * timeMultiple));
        flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		String error = influxDBConnection.deleteMeasurementData(sqlBuilder.toString());
		return StringUtil.isEmpty(error);
	}

	/**
	 * 根据rowKey删除指定的行
	 * @param pointNumber 点号
	 * @return boolean
	 */
	public boolean deleteAllRow(Long pointNumber,
                                String frequency) {
        String tableName = InfluxDBConsts.getTableName(frequency);
		String sql = String.format("delete from %s where pid='%s'",
                tableName, pointNumber);
		String error = influxDBConnection.deleteMeasurementData(sql);
		return StringUtil.isEmpty(error);
	}

	/**
	 * 获取dtime之后的所有数据
	 */
	public List<Measurement> getRowDataAfterDtime(Long pointNumber,
												  String dtime,
												  String frequency,
												  Integer flag) {
		if (pointNumber == null || dtime == null) {
			return null;
		}
		long rtime = getTimeCondition(dtime, pointNumber);
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder();
		if (!MeasurementConsts.FREQUENCY.Original.equals(frequency)) {
			sqlBuilder.append(String.format("select * from %s where pid='%s' and time>=%s",
					tableName, pointNumber, rtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		} else {
			sqlBuilder.append(String.format("select * from %s where pid='%s' and time<=%s",
					tableName, pointNumber, rtime * timeMultiple));
			flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
			sqlBuilder.append(" order by time desc");
		}
		return getQueryData(sqlBuilder.toString());
	}

	/**
	 * 根据条件获得一段时间内量测值最大的量测点
	 * @param pointNumbers 点号列表
	 * @param startDTime 开始时间
	 * @param endTDime 结束时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 */
	public List<Measurement> getMaxValueByTime(List<Long> pointNumbers,
											   String startDTime,
											   String endTDime,
											   String frequency,
											   Integer flag) {

		// 用第一个点号来区分时区
		Long pointNumber = pointNumbers.get(0);
		// 如果点号超过一定数量，会出现效率问题，所以采用分片形式
		if (pointNumbers.size() > searchShardSize) {
			List<Measurement> result = new ArrayList<>();
			int size = pointNumbers.size();
			int index = size / searchShardSize;
			for (int i = 0; i < index + 1; i++) {
				int subEnd = size;
				if (i < index) {
					subEnd = i * searchShardSize + searchShardSize;
				}
				List<Measurement> measurements = getMaxValueMilliByTime(pointNumbers.subList(i * searchShardSize, subEnd),
						getTimeCondition(startDTime, pointNumber), getTimeCondition(endTDime, pointNumber), frequency, flag);
				if (measurements != null) {
					result.addAll(measurements);
				}
			}
			return result;
		} else {
			return getMaxValueMilliByTime(pointNumbers, getTimeCondition(startDTime, pointNumber),
					getTimeCondition(endTDime, pointNumber), frequency, flag);
		}
	}

	/**
	 * 根据条件获得一段时间内量测值最大的量测点
	 * @param pointNumbers 点号列表
	 * @param startDTime 开始时间
	 * @param endTDime 结束时间
	 * @param frequency 时间频率
	 * @param flag 标记
	 */
	public List<Measurement> getMaxValueMilliByTime(List<Long> pointNumbers,
													Long startDTime,
													Long endTDime,
													String frequency,
													Integer flag) {
		Optional<Integer> flagOpt = Optional.ofNullable(flag);
		String tableName = InfluxDBConsts.getTableName(frequency);
		StringBuilder sqlBuilder = new StringBuilder(String.format("select time, pid, max(val) as val, flag from %s where time>=%s and time<=%s and %s",
				tableName, startDTime * timeMultiple, endTDime * timeMultiple, getPidCondition(pointNumbers)));
		flagOpt.ifPresent(e->sqlBuilder.append(String.format(" and flag=%s", e)));
		sqlBuilder.append(" group by pid");
		return getQueryData(sqlBuilder.toString());
	}


}
